package cn.com.cetccst.mqtt;

import android.os.Debug;

import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.HashMap;
import java.util.Map;

import cn.com.cetccst.mqtt.log.MqttLogger;

public final class MqttClientImp implements ISSMqttClient {

    private final MqttAsyncClient client;
    private final MqttManager.StateCallback stateCallback;
    private final Map<String, MqttManager.MessageCallback> topicCallbacks;
    private final static int QOS = 2;

    private final IMqttMessageListener messageListener;

    private MqttManager.STATE state = MqttManager.STATE.DISCONNECTED;

    public MqttClientImp(String clientId, String server, MqttManager.StateCallback stateCallback) throws MqttException {

        client = new MqttAsyncClient(server, clientId, new MemoryPersistence());
        this.stateCallback = stateCallback;
        topicCallbacks = new HashMap<>();
        messageListener = (topic, message) -> {
            MqttManager.MessageCallback messageCallback = topicCallbacks.get(topic);
            if (messageCallback != null)
                messageCallback.onMessageArrived(new String(message.getPayload()));
        };
        client.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectionLost(Throwable cause) {
                changeState(MqttManager.STATE.RECONNECTING);
            }

            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                for (String topic : topicCallbacks.keySet()) {
                    try {
                        client.subscribe(topic, QOS, messageListener);
                    } catch (MqttException e) {
                        MqttLogger.error("mqtt subscribe error", e);
                    }
                }
                changeState(MqttManager.STATE.CONNECTED);
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                MqttLogger.info("callback topic[%s], message[%s]", topic, new String(message.getPayload()));
                MqttManager.MessageCallback messageCallback = topicCallbacks.get(topic);
                if (messageCallback != null)
                    messageCallback.onMessageArrived(new String(message.getPayload()));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                MqttLogger.info("deliveryComplete");
                //TODO delivery 完成暂不处理
            }
        });
    }

    @Override
    public void connect() {
        if (state != MqttManager.STATE.DISCONNECTED) {
            return;
        }

        changeState(MqttManager.STATE.CONNECTING);
        MqttConnectOptions options = new MqttConnectOptions();
        options.setAutomaticReconnect(true);
        options.setCleanSession(false);
        options.setConnectionTimeout(30);
        options.setKeepAliveInterval(300); //seconds

        try {
            client.connect(options, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    MqttLogger.info("connect onSuccess");
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    MqttLogger.error("connect onFailure", exception);
                }
            });
        } catch (MqttException e) {
            MqttLogger.error("mqtt connect error", e);
            throw new RuntimeException(e);
        }
    }

    @Override
    public void disconnect() {
        try {
            client.disconnect();
            client.close();
            changeState(MqttManager.STATE.DISCONNECTED);
        } catch (MqttException e) {
            MqttLogger.error("mqtt disconnect or close error", e);
            throw new RuntimeException(e);
        }
    }

    private void changeState(MqttManager.STATE state) {
        this.state = state;
        if (state != null)
            stateCallback.onStateChanged(state);
    }

    @Override
    public boolean subscribeTopic(String topic, MqttManager.MessageCallback messageCallback) {
        topicCallbacks.put(topic, messageCallback);
        if (client.isConnected()) {
            try {
                client.subscribe(topic, QOS, messageListener);
                return true;
            } catch (MqttException e) {
                MqttLogger.error("subscribe Topic " + topic + "error", e);
            }
        }
        return false;
    }

    @Override
    public boolean unSubscribeTopic(String topic) {
        topicCallbacks.remove(topic);
        if (client.isConnected()) {
            try {
                client.unsubscribe(topic);
                return true;
            } catch (MqttException e) {
                MqttLogger.error("unSubscribe Topic " + topic + "error", e);
            }
        }
        return false;
    }
}
